package com.shujia.flink.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object Demo1DSToTable {
  def main(args: Array[String]): Unit = {
    //创建flink流处理环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //创建flink table环境
    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    /**
     * 1、构建一个流
     */
    val wordsDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 2、在流上定义表
     */
    tEnv.createTemporaryView("words", wordsDS, $"word")

    /**
     * 3、使用连续查询，查询动态表
     *
     * TableL 相当于spark中的DataFrame,可以使用DSl对数据做处理
     */
    val countTable: Table = tEnv.sqlQuery(
      """
        |select
        | word,
        | count(1) as c
        |from
        | words
        |group by
        | word
        |""".stripMargin)

    /**
     * 4、将结果表转换成流
     */

    countTable.print()

    env.execute()
  }

}
